package com.springboot.demo.config;

import com.rabbitmq.client.Channel;
import com.springboot.demo.exception.BusinessException;
import com.springboot.demo.service.RabbitmqService;
import com.springboot.demo.utils.Constant;
import com.springboot.demo.utils.ResponseResult;
import com.springboot.demo.utils.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.bridge.MessageHandler;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


/**
 * @author wubc683
 * @desc
 * @date 2021-07-22 17:37
 */
@Configuration
@Slf4j
public class RabbitConfig {

    @Autowired
    CachingConnectionFactory cachingConnectionFactory;
    @Autowired
    RabbitmqService rabbitmqService;

    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     " + "相关数据：" + correlationData);
                System.out.println("ConfirmCallback:     " + "确认情况：" + ack);
                System.out.println("ConfirmCallback:     " + "原因：" + cause);
//                log.error("rabbitmq消息确认失败，相关数据：{},确认清空：{},原因：{}",correlationData,ack,cause);
//                throw new BusinessException(ResponseResult.RABBITMQ_MSG_ACK_ERROR);
            }
        });

        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                System.out.println("ReturnCallback:     " + "消息：" + returnedMessage.getMessage());
                System.out.println("ReturnCallback:     " + "回应码：" + returnedMessage.getReplyCode());
                System.out.println("ReturnCallback:     " + "回应信息：" + returnedMessage.getReplyText());
                System.out.println("ReturnCallback:     " + "交换机：" + returnedMessage.getExchange());
                System.out.println("ReturnCallback:     " + "路由键：" + returnedMessage.getRoutingKey());
            }
        });
        return rabbitTemplate;
    }

    /**
     * 接受消息的监听，这个监听会接受消息队列TestDirectQueue的消息
     * 针对消费者配置
     *
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory);
        // 当前消费者数量 开启⼏个线程去处理数据⽀持运⾏时动态
        container.setConcurrentConsumers(100);
        // 最⼤消费者数量 消息堵塞太多的时候，会帮我⾃动扩展到我的最⼤消费者
        container.setMaxConcurrentConsumers(200);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
//        container.setQueueNames(Constant.Rabbitmq.TEST_QUEUE);
//        container.addQueueNames(Constant.Rabbitmq.TOPIC_SYSTEM_QUEUE);
//        container.addQueueNames(Constant.Rabbitmq.TOPIC_USER_QUEUE);
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
//        container.addQueues(new Queue("TestDirectQueue2",true));
        container.addQueues(new Queue(Constant.Rabbitmq.CLIENT_TO_WEB_QUEUE, true));

        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                try {
                    log.info("消费的主题消息来自：" + message.getMessageProperties().getConsumerQueue());
                    log.info("消息内容：" + new String(message.getBody()));
                    String consumerQueue = message.getMessageProperties().getConsumerQueue();
                    // 处理业务消息
                    Boolean consumeSuccess = rabbitmqService.processMessage(new String(message.getBody()));
                    // 同⼀时刻服务器只会发⼀条消息给消费者 多个消费者一人一条轮流分发
                    channel.basicQos(1000);
                    // 业务处理成功，手动确认
                    if (consumeSuccess = true) {
                        channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
                    }
                } catch (Exception e) {
                    //第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
                    channel.basicReject(deliveryTag, false);
                    log.error("rabbitmq消费端消费消息出错，消息内容：{},exception:{}", new String(message.getBody()), e);
                }
            }
        });
        return container;
    }
}